package com.xl.flinkdemo.processAPI;

import com.xl.flinkdemo.entity.SensorReading;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Tuple;

/**
 * @ClassName processApiDemo
 * @Description 实现一个温度监控，如果10s内温度连续上升，则进行报警
 * @Author hxl
 * @Date 2021/5/24 18:02
 * Version 1.0
 **/
public class processApiDemo {
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);

    DataStream<String> dataStream =
        env.readTextFile("D:\\studyspace\\flinkdemo\\src\\main\\resources\\sensors\\input.txt");

    DataStream<SensorReading> resultStream = dataStream.map(line -> {
      String[] fields = line.split(",");
      return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
    });

    //实现一个自定义的process
    resultStream.process(new MyProcessFunction());

    env.execute();
  }

  public static class MyProcessFunction extends ProcessFunction<SensorReading, Tuple2<Double, SensorReading>> {

    //定时器
    ValueState<Long> tsTimer;
    //上一个的温度值
    ValueState<Double> sensorValue;

    @Override
    public void open(Configuration parameters) throws Exception {
      //初始化状态
      tsTimer = getRuntimeContext().getState(new ValueStateDescriptor<Long>("ts-timer", Long.class));
      sensorValue = getRuntimeContext().getState(new ValueStateDescriptor<Double>("sensors", Double.class));
    }

    @Override
    public void processElement(SensorReading value, Context ctx, Collector<Tuple2<Double, SensorReading>> out)
        throws Exception {
      if (value.getSensors() - sensorValue.value() >= 0) {
        //开启定时器并更新温度状态
        sensorValue.update(value.getSensors());
        ctx.timerService().registerEventTimeTimer(value.getTimestamp());
        tsTimer.update(value.getTimestamp());
      } else {
    // 删除上一个闹钟
        ctx.timerService().deleteEventTimeTimer(value.getTimestamp());
        sensorValue.update(value.getSensors());
      }
    }

    // 每次触发时要做的
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<Double, SensorReading>> out)
        throws Exception {

    }
  }
}
